package org.ws.task.example.batch;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.ws.task.item.PullTask;
import org.ws.task.model.TaskItem;
import org.ws.task.model.protocol.CommitResponse;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Configuration
public class BatchJobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private PullTask<String> uuidPullTask;

    @Bean
    public Job uuidJob(){
        return this.jobBuilderFactory.get("uuidJob")
                .incrementer(new RunIdIncrementer())
                .start(uuidStep())
                .build();
    }

    @Bean
    public Step uuidStep(){
        return this.stepBuilderFactory.get("uuidStep")
                .<TaskItem<String>,TaskItem<String>>chunk(500)
                .reader(uuidItemReader())
                .processor(new ItemProcessor<TaskItem<String>, TaskItem<String>>() {
                    @Override
                    public TaskItem<String> process(TaskItem<String> taskItem) throws Exception {
                        return taskItem;
                    }
                })
                .writer(new ItemWriter<TaskItem<String>>() {
                    private AtomicInteger count = new AtomicInteger(0);
                    @Override
                    public void write(List<? extends TaskItem<String>> list) throws Exception {
                        if(CollectionUtils.isNotEmpty(list)){
                            uuidPullTask.commitSync(Lists.newArrayList(list), 3_000);
                            int i = count.addAndGet(list.size());
                            if(i/100 % 10 == 0 || i >= 500000){
                                log.info("process count:{}",i);
                            }
                        }
                    }
                })
                .faultTolerant()
                .retry(Exception.class)
                .skip(Exception.class)
                .skipLimit(999)
                .build();

    }

    @Bean
    public ItemReader<TaskItem<String>> uuidItemReader(){
        return new ItemReader<TaskItem<String>>() {
            @Override
            public TaskItem<String> read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                TaskItem<String> poll = uuidPullTask.poll();
                return poll;
            }
        };
    }

}
